package com.example.test_1_21.mqserver;

import com.example.test_1_21.common.Consumer;
import com.example.test_1_21.common.MQException;
import com.example.test_1_21.mqserver.core.*;
import com.example.test_1_21.mqserver.datacenter.DiskDataCenter;
import com.example.test_1_21.mqserver.datacenter.MemoryDataCenter;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

//通过这个类，来表示虚拟主机
//每个主机下面都管理自己的交换机，队列，绑定，消息
//同时提供api供上层调用
public class VirtualHost {
    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter=new MemoryDataCenter();
    private DiskDataCenter diskDataCenter=new DiskDataCenter();
    private Router router=new Router();
    private ConsumerManager consumerManager=new ConsumerManager(this);

    //操作交换机的锁对象
    private final Object exchangeLocker=new Object();

    //操作队列的锁对象
    private final Object queueLocker=new Object();

    public String getVirtualHostName() {
        return virtualHostName;
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public VirtualHost(String name){
        this.virtualHostName=name;

        //对于MemoryDataCenter来说，不需要额外的初始化操作的，只要对象new出来即可
        //但是针对DiskDataCenter来说，则需要进行初始化操作，建库建表和初始数据的设定
        diskDataCenter.init();

        //另外还需要针对硬盘的数据，进行恢复到内存中
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (MQException |IOException |ClassNotFoundException e) {
            e.printStackTrace();
            System.out.println("[VirtualHost] 恢复内存数据失败！");
        }
    }

    //创建交换机，如果不存在就创建，如果存在就直接返回
    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable, boolean autoDelete, Map<String,Object> arguments){
        //把交换机的名字加上虚拟主机作为前缀
        exchangeName=virtualHostName+exchangeName;
        try {
            synchronized (exchangeLocker) {
                //判定该交换机是否已经存在，直接通过内存查询
                Exchange existsExchange = memoryDataCenter.getExchange(exchangeName);
                if (existsExchange != null) {
                    System.out.println("[VirtualHost] 交换机已经存在！exchangeName=" + exchangeName);
                    return true;
                }
                //真正创建交换机，先构造Exchange对象
                Exchange exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);
                //把交换机对象写入硬盘
                if (durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                //把交换机对象写入对象
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建完成！exchangeName=" + exchangeName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 交换机创建失败！exchangeName="+exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //删除交换机
    public boolean exchangeDelete(String exchangeName) {
        exchangeName=virtualHostName+exchangeName;
        try {
            synchronized (exchangeLocker) {
                // 1. 先找到对应的交换机.
                Exchange toDelete = memoryDataCenter.getExchange(exchangeName);
                if (toDelete == null) {
                    throw new MQException("[VirtualHost] 交换机不存在无法删除!");
                }
                // 2. 删除硬盘上的数据
                if (toDelete.isDurable()) {
                    diskDataCenter.deleteExchange(exchangeName);
                }
                // 3. 删除内存中的交换机数据
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功! exchangeName=" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败! exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    //创建队列
    public boolean queueDeclare(String queueName,boolean durable,boolean exclusive,boolean autoDelete,Map<String,Object> arguments){
        queueName=virtualHostName+queueName;
        try {
            synchronized (exchangeLocker) {
                MSGQueue existsQueue = memoryDataCenter.getQueue(queueName);
                if (existsQueue != null) {
                    System.out.println("[VirtualHost] 队列已经存在！queueName=" + queueName);
                    return true;
                }
                MSGQueue queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);
                if (durable) {
                    diskDataCenter.insertQueue(queue);
                }
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功！queueName=" + queueName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 队列创建失败！queueName="+queueName);
            e.printStackTrace();
            return false;
        }
    }

    //删除队列
    public boolean queueDelete(String queueName){
        queueName=virtualHostName+queueName;
        try {
            synchronized (exchangeLocker) {
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if (queue == null) {
                    throw new MQException("[VirtualHost] 队列不存在！无法删除！queueName=" + queueName);
                }
                if (queue.isDurable()) {
                    diskDataCenter.deleteQueue(queueName);
                }
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 删除队列成功！queueName=" + queueName);
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 删除队列失败！queueName="+queueName);
            e.printStackTrace();
            return false;
        }
    }

    //创建绑定
    public boolean queueBind(String queueName,String exchangeName,String bindingKey){
        queueName=virtualHostName+queueName;
        exchangeName=virtualHostName+exchangeName;
        try {
            synchronized (exchangeLocker){
                synchronized (queueLocker){
                    Binding existsBinding=memoryDataCenter.getBinding(exchangeName,queueName);
                    if(existsBinding!=null){
                        throw new MQException("[VirtualHost] binding已经存在！queueName="+queueName+",exchangeName="+exchangeName);
                    }

                    if(router.checkBindingKey(bindingKey)){
                        System.out.println("[VirtualHost] bindingKey非法！bindingKey="+bindingKey);
                    }

                    Binding binding=new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);

                    MSGQueue queue=memoryDataCenter.getQueue(queueName);
                    if(queue==null){
                        throw new MQException("[VirtualHost] 队列不存在！queueName="+queueName);
                    }
                    Exchange exchange=memoryDataCenter.getExchange(exchangeName);
                    if(exchange==null){
                        throw new MQException("[VirtualHost] 交换机不存在！exchangeName="+exchangeName);
                    }

                    if(queue.isDurable()&&exchange.isDurable()){
                        diskDataCenter.insertBinding(binding);
                    }

                    memoryDataCenter.insertBinding(binding);
                }
            }
            System.out.println("[VirtualHost] 绑定创建成功！exchangeName="+exchangeName+",queueName="+queueName);
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 绑定创建失败！exchangeName="+exchangeName+",queueName="+queueName);
            e.printStackTrace();
            return false;
        }
    }

    //删除绑定
    public boolean queueUnbind(String queueName,String exchangeName){
        queueName=virtualHostName+queueName;
        exchangeName=virtualHostName+exchangeName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if (binding == null) {
                        throw new MQException("[VirtualHost] 删除绑定失败！绑定不存在！exchangeName" + exchangeName + ",queueName=" + queueName);
                    }

                    diskDataCenter.deleteBinding(binding);

                    memoryDataCenter.deleteBinding(binding);
                }
            }
            System.out.println("[VirtualHost] 删除绑定成功！");
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 删除绑定失败！");
            e.printStackTrace();
            return false;
        }
    }

    //发送消息到指定的交换机/队列中
    public boolean basicPublish(String exchangeName,String routingKey,BasicProperties basicProperties,byte[] body){
        try{
            //1、转换交换机的名字
            exchangeName=virtualHostName+exchangeName;
            //2、检查routingKey是否合法
            if(!router.checkRoutingKey(routingKey)){
                throw new MQException("[VirtualHost] routingKey非法！routingKey="+routingKey);
            }
            //3、查找交换机对象
            Exchange exchange=memoryDataCenter.getExchange(exchangeName);
            if(exchange==null){
                throw new MQException("[VirtualHost] 交换机不存在！exchangeName="+exchangeName);
            }
            //4、判断交换机类型
            if(exchange.getType()==ExchangeType.DIRECT){
                //按照直接交换机的方式
                String queueName=virtualHostName+routingKey;
                //5、构造消息对象
                Message message=Message.createMessageWithId(routingKey,basicProperties,body);
                //6、查找该队列名对应的对象
                MSGQueue queue=memoryDataCenter.getQueue(queueName);
                if(queue==null){
                    throw new MQException("[VirtualHost] 队列不存在！queueName="+queueName);
                }
                //7、队列存在，直接给队列中写入消息
                sendMessage(queue,message);
            }else{
                //按照fanout和topic方式
                ConcurrentHashMap<String,Binding> bindingsMap=memoryDataCenter.getBindings(exchangeName);
                for (Map.Entry<String,Binding> entry:bindingsMap.entrySet()){
                    //获取绑定对象，判断队列是否存在
                    Binding binding=entry.getValue();
                    MSGQueue queue=memoryDataCenter.getQueue(binding.getQueueName());
                    if(queue==null){
                        System.out.println("[VirtualHost] basicPublish发送消息时，发现队列不存在！queueName="+binding.getQueueName());
                        continue;
                    }
                    //构造一个消息对象
                    Message message=Message.createMessageWithId(routingKey,basicProperties,body);
                    //判断该消息能否转发给该队列
                    if(!router.route(exchange.getType(),binding,message)){
                        continue;
                    }
                    //真正转发消息给队列
                    sendMessage(queue,message);
                }
            }
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] 消息发送失败！");
            e.printStackTrace();
            return false;
        }
    }

    private void sendMessage(MSGQueue queue, Message message) throws MQException, IOException, InterruptedException {
        int deliverMode=message.getDeliverMode();
        //deliverMode为1，不持久化；为2，表示持久化
        if(deliverMode==2){
            diskDataCenter.sendMessage(queue,message);
        }
        //写入内存
        memoryDataCenter.sendMessage(queue,message);
        consumerManager.notifyConsume(queue.getName());
    }

    //订阅消息
    //添加一个队列的订阅者，当队列收到消息后，就要把消息推送给对应的订阅者
    //consumerTag：消费者的身份标识
    //autoAck：消息被消费完成，应答的方式，为true自动应答，为false手动应答
    //consumer：是一个回调函数，此处类型设定成函数式接口，这样后续调用basicConsume并且传实参时就可以写作lambda表达式
    public boolean basicConsume(String consumerTag,String queueName,boolean autoAck, Consumer consumer){
        //构造一个ConsumerEnv对象，把这个对应的队列找到，再把这个Consumer对象添加到该队列中
        queueName=virtualHostName+queueName;
        try {
            consumerManager.addConsumer(consumerTag,queueName,autoAck,consumer);
            System.out.println("[VirtualHost] basicConsume成功！queueName="+queueName);
        }catch (Exception e){
            System.out.println("[VirtualHost] basicConsume失败！queueName="+queueName);
            e.printStackTrace();
            return false;
        }
        return true;
    }

    public boolean basicAck(String queueName,String messageId){
        queueName=virtualHostName+queueName;
        try {
            //1、获取到消息的队列
            Message message=memoryDataCenter.getMessage(messageId);
            if(message==null){
                throw new MQException("[VirtualHost] 要确认的消息不存在！messageId="+messageId);
            }
            MSGQueue queue=memoryDataCenter.getQueue(queueName);
            if(queue==null){
                throw new MQException("[VirtualHost] 要确认的队列不存在！queueName="+queueName);
            }
            //2、删除硬盘上的数据
            if(message.getDeliverMode()==2){
                diskDataCenter.deleteMessage(queue,message);
            }
            //3、删除消息中心的数据
            memoryDataCenter.removeMessage(messageId);
            //4、删除待确认的集合中的数据
            memoryDataCenter.removeMessageWaitAck(queueName,messageId);
            System.out.println("[VirtualHost] basicAck成功！消息被成功确认！queueName="+queueName+",messageId="+messageId);
            return true;
        }catch (Exception e){
            System.out.println("[VirtualHost] basicAck失败！消息确认失败！queueName="+queueName+",messageId="+messageId);
            return false;
        }
    }
}
